package net.bw.realtime.warehouse.dwd;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bw.realtime.warehouse.dwd
 * @Author: zhangleqing
 * @CreateTime: 2025-08-25  20:31
 * @Description: TODO
 * @Version: 1.0
 */
public class Cars {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


/*
CREATE TABLE ods_cars(
    `id` BIGINT,
    `owerId` BIGINT,
    `carCode` STRING,
    `carColor` STRING,
    `type` BIGINT,
    `remark` STRING,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc', -- 使用 MySQL CDC 连接器
    'hostname' = 'node101', -- MySQL 服务器主机名
    'port' = '3306', -- MySQL 服务端口
    'username' = 'root', -- 数据库用户名
    'password' = '123456', -- 数据库密码
    'database-name' = 'sca', -- 数据库名称
    'table-name' = 'cars', -- 源表名称
    'connect.timeout' = '60s' -- 连接超时时间（60秒）
)
 */
        // 1.数据读取
        tEnv.executeSql(
                "CREATE TABLE ods_cars(\n" +
                        " id BIGINT,\n" +
                        " owerId BIGINT,\n" +
                        " carCode STRING,\n" +
                        " carColor STRING,\n" +
                        " type STRING,\n" +
                        " remark STRING,\n" +
                        " PRIMARY KEY (id) NOT ENFORCED\n" +
                        ")  WITH (\n" +
                        " 'connector' = 'mysql-cdc',\n" +
                        " 'hostname' = 'node101',\n" +
                        " 'port' = '3306',\n" +
                        " 'username' = 'root',\n" +
                        " 'password' = '123456',\n" +
                        " 'database-name' = 'sca',\n" +
                        " 'table-name' = 'cars',\n" +
                        " 'connect.timeout' = '60s'\n" +
                        ")"
        );

/*
create table ods_cars_clean(
    `id` BIGINT,
    `owerId` BIGINT,
    `carCode` STRING,
    `carColor` STRING,
    `type` STRING,
    `remark` STRING,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'ods_cars_clean',
    'properties.bootstrap.servers' = 'node101:9092',
    'key.format' = 'json',
    'value.format' = 'json',
    'sink.delivery-guarantee' = 'exactly-once',
    'properties.transaction.timeout.ms' = '900000',
    'sink.parallelism' = '3'
)

1. 'sink.delivery-guarantee' = 'exactly-once'
作用：确保端到端精确一次语义
解决的问题：防止数据重复或丢失

2. 'properties.transaction.timeout.ms' = '900000'
作用：设置 Kafka 事务超时时间
推荐值：900000ms = 15分钟

3. 'sink.parallelism' = '3'
作用：设置 Sink 算子并行度
 */
        // 2.数据写入
        tEnv.executeSql(
                "create table ods_cars_clean(\n" +
                        "    id BIGINT,\n" +
                        "    owerId BIGINT,\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    type STRING,\n" +
                        "    remark STRING,\n" +
                        "    PRIMARY KEY (id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'ods_cars_clean',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json',\n" +
                        "    'sink.delivery-guarantee' = 'exactly-once',\n" +
                        "    'properties.transaction.timeout.ms' = '900000',\n" +
                        "    'sink.parallelism' = '3'\n" +
                        ")"
        );

/*
insert into ods_cars_clean
select
    id,
    owerId,
    carCode,
    carColor,
    CASE
        WHEN type = '1' THEN '大型'
        WHEN type = '2' THEN '中型'
        WHEN type = '3' THEN '小型'
        ELSE '未知类型'
    END AS carType,
    remark
from ods_cars
where id is not null
and owerId is not null
 */
        // 3.数据清洗插入
        tEnv.executeSql(
                "insert into ods_cars_clean\n" +
                        "select\n" +
                        "    id,\n" +
                        "    owerId,\n" +
                        "    carCode,\n" +
                        "    carColor,\n" +
                        "    CASE\n" +
                        "        WHEN type = '1' THEN '大型'\n" +
                        "        WHEN type = '2' THEN '中型'\n" +
                        "        WHEN type = '3' THEN '小型'\n" +
                        "        ELSE '未知类型'\n" +
                        "    END AS type,\n" +
                        "    remark\n" +
                        "from ods_cars\n" +
                        "where id is not null\n" +
                        "and owerId is not null"
        );


        env.execute("Cars");
    }
}
